// Copyright (c) 2020-present, INSPUR Co, Ltd. All rights reserved.
// This source code is licensed under Apache 2.0 License.

#include <ctime>
#include <memory>
#include <string>
#include <thread>

#include "db/db_test_util.h"
#include "db/memtable.h"
#include "db/range_del_aggregator.h"
#include "port/stack_trace.h"
#include "rangearena/range_arena.h"
#include "rocksdb/memtablerep.h"
#include "rocksdb/slice_transform.h"

namespace rocksdb {

class DBMemTableTest : public DBTestBase {
 public:
  DBMemTableTest() : DBTestBase("/db_memtable_test") {}

  std::string Get2(const Slice& k) {
    ReadOptions options;
    options.verify_checksums = true;
    std::string result;
    Status s = db_->Get(options, k, &result);
    if (s.IsNotFound()) {
      result = "NOT_FOUND";
    } else if (!s.ok()) {
      result = s.ToString();
    }
    return result;
  }
};

class MockMemTableRep : public MemTableRep {
 public:
  explicit MockMemTableRep(Allocator* allocator, MemTableRep* rep)
      : MemTableRep(allocator), rep_(rep), num_insert_with_hint_(0) {}

  KeyHandle AllocatePure(const size_t len, char** buf, const Slice& key,
                         void** rangearena) override {
    return rep_->AllocatePure(len, buf, key, rangearena);
  }

  void Insert(KeyHandle handle) override { rep_->Insert(handle); }

  void InsertWithHint(KeyHandle handle, void** hint) override {
    num_insert_with_hint_++;
    EXPECT_NE(nullptr, hint);
    last_hint_in_ = *hint;
    rep_->InsertWithHint(handle, hint);
    last_hint_out_ = *hint;
  }

  bool Contains(const char* key) const override { return rep_->Contains(key); }

  void Get(const LookupKey& k, void* callback_args,
           bool (*callback_func)(void* arg, const char* entry)) override {
    rep_->Get(k, callback_args, callback_func);
  }

  size_t ApproximateMemoryUsage() override {
    return rep_->ApproximateMemoryUsage();
  }

  Iterator* GetIterator(Arena* arena) override {
    return rep_->GetIterator(arena);
  }

  void* last_hint_in() { return last_hint_in_; }
  void* last_hint_out() { return last_hint_out_; }
  int num_insert_with_hint() { return num_insert_with_hint_; }

 private:
  std::unique_ptr<MemTableRep> rep_;
  void* last_hint_in_;
  void* last_hint_out_;
  int num_insert_with_hint_;
};

class MockMemTableRepFactory : public MemTableRepFactory {
 public:
  MemTableRep* CreateMemTableRep(const MemTableRep::KeyComparator& cmp,
                                 Allocator* allocator,
                                 const SliceTransform* transform,
                                 Logger* logger) override {
    PureMemFactory factory;
    MemTableRep* skiplist_rep =
        factory.CreateMemTableRep(cmp, allocator, transform, logger);
    mock_rep_ = new MockMemTableRep(allocator, skiplist_rep);
    return mock_rep_;
  }

  MemTableRep* CreateMemTableRep(const MemTableRep::KeyComparator& cmp,
                                 Allocator* allocator,
                                 const SliceTransform* transform,
                                 Logger* logger,
                                 uint32_t column_family_id) override {
    last_column_family_id_ = column_family_id;
    return CreateMemTableRep(cmp, allocator, transform, logger);
  }

  const char* Name() const override { return "MockMemTableRepFactory"; }

  MockMemTableRep* rep() { return mock_rep_; }

  bool IsInsertConcurrentlySupported() const override { return false; }

  uint32_t GetLastColumnFamilyId() { return last_column_family_id_; }

 private:
  MockMemTableRep* mock_rep_;
  // workaround since there's no port::kMaxUint32 yet.
  uint32_t last_column_family_id_ = static_cast<uint32_t>(-1);
};

class TestPrefixExtractor : public SliceTransform {
 public:
  const char* Name() const override { return "TestPrefixExtractor"; }

  Slice Transform(const Slice& key) const override {
    const char* p = separator(key);
    if (p == nullptr) {
      return Slice();
    }
    return Slice(key.data(), p - key.data() + 1);
  }

  bool InDomain(const Slice& key) const override {
    return separator(key) != nullptr;
  }

  bool InRange(const Slice& /*key*/) const override { return false; }

 private:
  const char* separator(const Slice& key) const {
    return reinterpret_cast<const char*>(memchr(key.data(), '_', key.size()));
  }
};

Slice genMVCCKey(char* key) {
  uint32_t keySize = VarintLength(strlen(key)) + strlen(key) + 1;
  char* ret = new char[keySize];
  memset(ret, '\0', keySize);
  char* p = EncodeVarint32(ret, strlen(key));
  memcpy(p, key, strlen(key));
  return Slice(ret, keySize);
}

Slice genMVCCKey1(const char* key) {
  uint32_t keySize = strlen(key) + 1;
  char* ret = new char[keySize];
  memset(ret, '\0', keySize);
  memcpy(ret, key, strlen(key));
  return Slice(ret, keySize);
}



char *randstr(char *str, const int len)
{

  int i;
  for (i = 0; i < len; ++i)
  {
    switch ((rand() % 3))
    {
      case 1:
        str[i] = 'A' + rand() % 26;
        break;
      case 2:
        str[i] = 'a' + rand() % 26;
        break;
      default:
        str[i] = '0' + rand() % 10;
        break;
    }
  }
  str[++i] = '\0';
  return str;
}

TEST_F(DBMemTableTest, HashIndexPerformance) {
  Options options = CurrentOptions();
  // Create a MemTable
  InternalKeyComparator cmp(BytewiseComparator());
  options.memtable_factory = std::make_shared<PureMemFactory>();
  options.pureMemTable = true;
  options.cf_paths.emplace_back(dbname_, 0);
  ImmutableCFOptions iOptions(options);
  WriteBufferManager wb(options.db_write_buffer_size);
  MemTable* mem = new MemTable(cmp, iOptions, MutableCFOptions(options), &wb,
                               kMaxSequenceNumber, 0 /* column_family_id */);
  std::cout << "memtable create ok." << std::endl;
  // Write some keys and make sure it returns false on duplicates
  bool res;
  std::cout << "进入循环" << std::endl;


  std::vector<std::string> keys;

  int N = 1000000;
  srand(time(NULL));
  for(int i=0; i<N; i++){
    char key[20];
    keys.push_back(randstr(key, 14));
  }

  clock_t startTime,addTime, getTime;
  char tmp[1000] = {'0'};
  SequenceNumber seq = 0;
  startTime = clock();
  for(int i=0; i<N; i++){
    seq++;
    ASSERT_TRUE(mem->Add(seq, kTypeValue, genMVCCKey1(keys[i].data()), Slice(tmp, 104)));
    seq++;
    ASSERT_TRUE(mem->Add(seq, kTypeValue, genMVCCKey1(keys[i].data()), Slice(tmp, 105)));
  }

  addTime = clock();

  std::cout << "addTime: " <<(double)(addTime - startTime)*1000 / CLOCKS_PER_SEC << "ms" << std::endl;

  std::string value;
  Status s;
  SequenceNumber sq = 0;
  seq = 0;
  ReadOptions read_opts;
  MergeContext merge_context;

  for(int i=0; i<N; i++){
    seq++;
    res = mem->Get(LookupKey(genMVCCKey1(keys[i].data()), seq), &value, &s,
                   &merge_context, &sq, &sq, read_opts);
    ASSERT_TRUE(res);
    ASSERT_EQ(value.length(), 104);
    seq++;

    res = mem->Get(LookupKey(genMVCCKey1(keys[i].data()), seq), &value, &s,
                   &merge_context, &sq, &sq, read_opts);
    ASSERT_TRUE(res);
    ASSERT_EQ(value.length(), 105);
  }

  getTime = clock();
  std::cout << "getTime: " <<(double)(getTime - addTime)*1000 / CLOCKS_PER_SEC << "ms" << std::endl;


  std::cout << "delete mem" << std::endl;
  delete mem;
  std::cout << "-------------------------------------------------------end ..."
            << std::endl;
}

int parseLine(char *line) {
  // This assumes that a digit will be found and the line ends in " Kb".
  int i = strlen(line);
  const char *p = line;
  while (*p < '0' || *p > '9') p++;
  line[i - 3] = '\0';
  i = atoi(p);
  return i;
}

// 虚拟内存和物理内存，单位为kb
typedef struct {
  uint32_t virtualMem;
  uint32_t physicalMem;
} processMem_t;

processMem_t GetProcessMemory() {
  FILE *file = fopen("/proc/self/status", "r");
  char line[128];
  processMem_t processMem;

  while (fgets(line, 128, file) != NULL) {
    if (strncmp(line, "VmSize:", 7) == 0) {
      processMem.virtualMem = parseLine(line);
      break;
    }

    if (strncmp(line, "VmRSS:", 6) == 0) {
      processMem.physicalMem = parseLine(line);
      break;
    }
  }
  fclose(file);
  return processMem;
}

TEST_F(DBMemTableTest, memRecover) {
  std::string sst_data;
  sst_data.reserve(48 << 20);
  for (int i =0; i<30000; i++){
    std::string key = "key" + ToString(i);
    std::string value(1000, 'v');
    SequenceNumber seq = i + 1;
    uint32_t key_size = static_cast<uint32_t>(key.size());
    uint32_t val_size = static_cast<uint32_t>(value.size());
    uint32_t internal_key_size = key_size + 8;
    const uint32_t encoded_len = VarintLength(internal_key_size) +
        internal_key_size + VarintLength(val_size) + val_size;

    char tmp_data[encoded_len];
    char* buf = tmp_data;
    char* p = EncodeVarint32(buf, internal_key_size);
    memcpy(p, key.data(), key_size);
    Slice key_slice(p, key_size);
    p += key_size;
    uint64_t packed = PackSequenceAndType(i, kTypeValue);
    EncodeFixed64(p, packed);
    p += 8;
    p = EncodeVarint32(p, val_size);
    memcpy(p, value.data(), val_size);
    sst_data += std::string(tmp_data, encoded_len);
//    std::cout << std::string(tmp_data, encoded_len) << std::endl;
  }
  auto data_ptr = sst_data.data();
  auto data_size = sst_data.size();
//  std::cout << data_ptr << std::endl;
//  std::cout << data_size << std::endl;
}

TEST_F(DBMemTableTest, flushManager) {
  Options options;
  // Create a MemTable
  InternalKeyComparator cmp(BytewiseComparator());
  options.memtable_factory = std::make_shared<PureMemFactory>();
  options.pureMemTable = true;
  options.cf_paths.emplace_back(dbname_, 0);
  options.cf_paths.emplace_back(dbname_, 1);
  ImmutableCFOptions ioptions(options);
  WriteBufferManager wb(options.db_write_buffer_size);
  auto mem = new MemTable(cmp, ioptions, MutableCFOptions(options), &wb,
                          kMaxSequenceNumber, 1 /* column_family_id */);

  std::cout << "memtable create ok." << std::endl;
  // Write some keys and make sure it returns false on duplicates
  bool res;
  std::cout << "Enter the loop" << std::endl;
  // Test the duplicate keys under stress
  char tmp[1000];
  memset(tmp, 70, 1000);
  SequenceNumber seq = 1000000;
  int record_sum = 100000;
  for (int i = 0; i < record_sum; i++) {
    seq++;
    std::string a = std::to_string(seq) + "_test";
    res = mem->Add(seq, kTypeValue, genMVCCKey1(a.data()), Slice(tmp, 1000));
    ASSERT_TRUE(res);
  }
  std::cout << "-----------insert all records ok." << std::endl;
  delete mem;
  std::cout << "-------------------------------------------------------end ..."
            << std::endl;
}

TEST_F(DBMemTableTest, IntervalFlushData) {
  Options options;
  // Create a MemTable
  InternalKeyComparator cmp(BytewiseComparator());
  options.memtable_factory = std::make_shared<PureMemFactory>();
  options.pureMemTable = true;
  options.cf_paths.emplace_back(dbname_, 0);
  options.cf_paths.emplace_back(dbname_, 1);
  ImmutableCFOptions ioptions(options);
  WriteBufferManager wb(options.db_write_buffer_size);
  auto mem = new MemTable(cmp, ioptions, MutableCFOptions(options), &wb,
                          kMaxSequenceNumber, 1 /* column_family_id */);

  std::cout << "memtable create ok." << std::endl;
  // Write some keys and make sure it returns false on duplicates
  bool res;
  std::cout << "begin insert: "<< std::endl;
  // Test the duplicate keys under stress
  char tmp[1000];
  memset(tmp, 70, 1000);
  SequenceNumber seq = 1000000000;
  int record_sum = 1000000;
  clock_t cstart,cends;
  cstart=clock();
  for (int i = 0; i < record_sum; i++) {
    seq++;
    std::string a = std::to_string(seq) + "_test";
    res = mem->Add(seq, kTypeValue, genMVCCKey1(a.data()), Slice(tmp, 1000));
    ASSERT_TRUE(res);
  }
  cends=clock();
  std::cout << "insert "<< record_sum << ", cost(ms): " << cends-cstart << std::endl;

  delete mem;
  std::cout << "-------------------------------------------------------end ..."
            << std::endl;
}

std::mutex flush_mu;
void multiFlushDataTest(MemTable* mem) {
  // wzk: number of thread
  static int threads_id = 10;
  flush_mu.lock();
  threads_id++;
  int thread_id = threads_id;
  flush_mu.unlock();

  char tmp[1000];
  memset(tmp, 70, 1000);
  SequenceNumber seq = thread_id * 1000000;
  int record_sum = 100000;
  for (int k = 0; k < record_sum; ++k) {
    //    std::this_thread::sleep_for(std::chrono::nanoseconds(1));
    seq++;
    std::string a = std::to_string(seq) + "_test";
    bool res =
        mem->Add(seq, kTypeValue, genMVCCKey1(a.data()), Slice(tmp, 1000));
    ASSERT_TRUE(res);
  };
}

TEST_F(DBMemTableTest, multiflushManager) {
  Options options;
  // Create a MemTable
  InternalKeyComparator cmp(BytewiseComparator());
  options.memtable_factory = std::make_shared<PureMemFactory>();
  options.pureMemTable = true;
  options.cf_paths.emplace_back(dbname_, 0);
  ImmutableCFOptions ioptions(options);
  WriteBufferManager wb(options.db_write_buffer_size);
  MemTable* mem = new MemTable(cmp, ioptions, MutableCFOptions(options), &wb,
                               kMaxSequenceNumber, 0 /* column_family_id */);

  std::cout << "memtable create ok." << std::endl;
  // Write some keys and make sure it returns false on duplicates
  bool res;
  std::cout << "Enter the loop" << std::endl;
  // Test the duplicate keys under stress

  std::thread threads[20];
  for (auto& t : threads) {
    t = std::thread(multiFlushDataTest, mem);
  }

  for (auto& t : threads) {
    t.join();
  }

  ReadOptions read_options;
  read_options.total_order_seek = true;
  InternalIterator* iter = mem->NewIterator(read_options, new Arena());
  iter->SeekToFirst();
  size_t totalKVNum = 0;
  while(iter->Valid()){
    totalKVNum ++;
    iter->Next();
  }

  std::cout << "-----------insert all records["<< totalKVNum << "] ok." << std::endl;
  delete mem;
  std::cout << "-------------------------------------------------------end ..."
            << std::endl;
}

TEST_F(DBMemTableTest, Expand) {
  Options options = CurrentOptions();
  // Create a MemTable
  InternalKeyComparator cmp(BytewiseComparator());
  options.memtable_factory = std::make_shared<PureMemFactory>();
  options.pureMemTable = true;
  options.cf_paths.emplace_back(dbname_, 0);
  ImmutableCFOptions ioptions(options);
  WriteBufferManager wb(options.db_write_buffer_size);
  MemTable* mem = new MemTable(cmp, ioptions, MutableCFOptions(options), &wb,
                               kMaxSequenceNumber, 0 /* column_family_id */);
  std::cout << "memtable create ok." << std::endl;
  // Write some keys and make sure it returns false on duplicates
  bool res;
  std::cout << "Enter the loop" << std::endl;
  // Test the duplicate keys under stress
  char tmp[1000] = {'0'};
  SequenceNumber seq = 900000;
  int record_sum = 64000;
  for (int i = 0; i < record_sum; i++) {
    seq++;
    std::string a = std::to_string(seq) + "_test";
    // std::cout << "a : " << a << ", a.data()" << a.data() << ", a.length:" <<
    // a.length()<< std::endl;
    res = mem->Add(seq, kTypeValue, genMVCCKey1(a.data()), Slice(tmp, 1000));
    ASSERT_TRUE(res);
    std::string value;
    Status s;
    ReadOptions read_opts;
    SequenceNumber sq = 0;
    MergeContext merge_context;
    res = mem->Get(LookupKey(genMVCCKey1(a.data()), seq), &value, &s,
                   &merge_context, &sq, &sq, read_opts);
    ASSERT_TRUE(res);
    ASSERT_EQ(value.length(), 1000);
  }
  std::cout << "-----------insert all records ok." << std::endl;
  // std::this_thread::sleep_for(std::chrono::milliseconds(1000));

  seq = 900000;
  for (int i = 0; i < record_sum; i++) {
    seq++;
    std::string a = std::to_string(seq) + "_test";
    std::string value;
    Status s;
    ReadOptions read_opts;
    SequenceNumber sq = 0;
    MergeContext merge_context;
    res = mem->Get(LookupKey(genMVCCKey1(a.data()), seq), &value, &s,
                   &merge_context, &sq, &sq, read_opts);
    ASSERT_TRUE(res);
    ASSERT_EQ(value.length(), 1000);
  }

  std::cout << "delete mem" << std::endl;
  delete mem;
  std::cout << "-------------------------------------------------------end ..."
            << std::endl;
}

TEST_F(DBMemTableTest, HugeDataInsert) {
  Options options;
  // Create a MemTable
  InternalKeyComparator cmp(BytewiseComparator());
  options.memtable_factory = std::make_shared<PureMemFactory>();
  options.pureMemTable = true;
  options.cf_paths.emplace_back(dbname_, 0);
  ImmutableCFOptions ioptions(options);
  WriteBufferManager wb(options.db_write_buffer_size);
  MemTable* mem = new MemTable(cmp, ioptions, MutableCFOptions(options), &wb,
                               kMaxSequenceNumber, 0 /* column_family_id */);
  std::cout << "memtable create ok." << std::endl;
  // Write some keys and make sure it returns false on duplicates
  bool res;
  std::cout << "Enter the loop" << std::endl;
  // Test the duplicate keys under stress
  char tmp[1000];
  memset(tmp, 255, 1000);
  SequenceNumber seq = 100000;
  int record_sum = 150000;
  for (int i = 0; i < record_sum; i++) {
    seq++;
    std::string a = std::to_string(seq) + "_test";
    res = mem->Add(seq, kTypeValue, genMVCCKey1(a.data()), Slice(tmp, 1000));
    ASSERT_TRUE(res);
  }

  std::cout << "-----------insert all records ok." << std::endl;
  // std::this_thread::sleep_for(std::chrono::milliseconds(5000));

  seq = 100000;
  for (int i = 0; i < record_sum; i++) {
    seq++;
    std::string a = std::to_string(seq) + "_test";
    std::string value;
    Status s;
    ReadOptions read_opts;
    SequenceNumber sq = 0;
    MergeContext merge_context;
    res = mem->Get(LookupKey(genMVCCKey1(a.data()), seq), &value, &s,
                   &merge_context, &sq, &sq, read_opts);
    ASSERT_TRUE(res);
    ASSERT_EQ(value.length(), 1000);
  }

  delete mem;
  std::cout << "-------------------------------------------------------end ..."
            << std::endl;
}

// RandomDataInsert
TEST_F(DBMemTableTest, RandomDataInsert) {
  Options options;
  // Create a MemTable
  InternalKeyComparator cmp(BytewiseComparator());
  options.memtable_factory = std::make_shared<PureMemFactory>();
  options.pureMemTable = true;
  options.cf_paths.emplace_back(dbname_, 0);
  ImmutableCFOptions ioptions(options);
  WriteBufferManager wb(options.db_write_buffer_size);
  MemTable* mem = new MemTable(cmp, ioptions, MutableCFOptions(options), &wb,
                               kMaxSequenceNumber, 0 /* column_family_id */);
  std::cout << "memtable create ok." << std::endl;
  // Write some keys and make sure it returns false on duplicates
  bool res;
  std::cout << "Enter the loop" << std::endl;
  // Test the duplicate keys under stress
  char tmp[1000];
  memset(tmp, 255, 1000);
  srand(time(NULL));
  SequenceNumber seq = 0;
  int record_sum = 5000;
  for (int i = 0; i < record_sum; i++) {
    int key = rand() % 10 + 10;
    seq = seq + key;
    std::string a = std::to_string(seq) + "_test";
    res = mem->Add(seq, kTypeValue, genMVCCKey1(a.data()), Slice(tmp, 1000));
    ASSERT_TRUE(res);

    std::string value;
    Status s;
    ReadOptions read_opts;
    SequenceNumber sq = 0;
    MergeContext merge_context;
    res = mem->Get(LookupKey(genMVCCKey1(a.data()), seq), &value, &s,
                   &merge_context, &sq, &sq, read_opts);
    ASSERT_TRUE(res);
    ASSERT_EQ(value.length(), 1000);
  }

  std::cout << "-----------insert all records ok." << std::endl;

  delete mem;
  std::cout << "-------------------------------------------------------end ..."
            << std::endl;
}


void insertAndGet(std::string str, MemTable* mem) {
  bool res;
  std::cout << "Enter the loop" << std::endl;
  // Test the duplicate keys under stress
  char tmp[1000];
  memset(tmp, 255, 1000);
  SequenceNumber seq = 1000000;
  int record_sum = 500000;
  for (int i = 0; i < record_sum; i++) {
    seq++;
    std::string a = std::to_string(seq) + str;
    res = mem->Add(seq, kTypeValue, genMVCCKey1(a.data()), Slice(tmp, 1000));
    ASSERT_TRUE(res);
  }

  std::cout << "-----------insert all this thread records ok." << std::endl;
}
// Insert data for multiple threads
TEST_F(DBMemTableTest, MultithreadInsert) {
  Options options;
  // Create a MemTable
  InternalKeyComparator cmp(BytewiseComparator());
  options.memtable_factory = std::make_shared<PureMemFactory>();
  options.pureMemTable = true;
  options.cf_paths.emplace_back(dbname_, 0);
  ImmutableCFOptions ioptions(options);
  WriteBufferManager wb(options.db_write_buffer_size);
  MemTable* mem = new MemTable(cmp, ioptions, MutableCFOptions(options), &wb,
                               kMaxSequenceNumber, 0 /* column_family_id */);

  std::thread test1(insertAndGet, "test", mem);
  std::thread test2(insertAndGet, "_tast", mem);
  std::thread test3(insertAndGet, "_ttst1", mem);
  std::thread test4(insertAndGet, "_ttst2", mem);
  std::thread test5(insertAndGet, "_ttst3", mem);
  std::thread test6(insertAndGet, "_ttst4", mem);
  std::thread test7(insertAndGet, "_ttst5", mem);
  std::thread test8(insertAndGet, "_ttst6", mem);
  //      insertAndGet("test");
  //      insertAndGet("tast");
  //      insertAndGet("twst");
  test1.join();
  test2.join();
  test3.join();
  test4.join();
  test5.join();
  test6.join();
  test7.join();
  test8.join();

  std::this_thread::sleep_for(std::chrono::seconds(5));

  ReadOptions read_options;
  read_options.total_order_seek = true;
  InternalIterator* iter = mem->NewIterator(read_options, new Arena());
  iter->SeekToFirst();
  size_t totalKVNum = 0;
  while(iter->Valid()){
    totalKVNum ++;
    iter->Next();
  }
  std::cout << "-----------insert all records["<< totalKVNum << "] ok." << std::endl;

  delete mem;
  std::cout << "-------------------------------------------------------end ..."
            << std::endl;
}

// The first step is to change the Manager's internal disassembly and expansion
// event to direct call processing logic.
TEST_F(DBMemTableTest, HugeDataInsertWithoutThread) {
  Options options;
  // Create a MemTable
  InternalKeyComparator cmp(BytewiseComparator());
  options.memtable_factory = std::make_shared<PureMemFactory>();
  options.pureMemTable = true;
  options.cf_paths.emplace_back(dbname_, 0);
  ImmutableCFOptions ioptions(options);
  WriteBufferManager wb(options.db_write_buffer_size);
  MemTable* mem = new MemTable(cmp, ioptions, MutableCFOptions(options), &wb,
                               kMaxSequenceNumber, 0 /* column_family_id */);
  std::cout << "memtable create ok." << std::endl;
  // Write some keys and make sure it returns false on duplicates
  bool res;
  std::cout << "Enter the loop" << std::endl;
  // Test the duplicate keys under stress
  char tmp[1000];
  memset(tmp, 255, 1000);
  SequenceNumber seq = 10000000;
  int record_sum = 300000;
  for (int i = 0; i < record_sum; i++) {
    seq++;
    std::string a = std::to_string(seq) + "_test";
    res = mem->Add(seq, kTypeValue, genMVCCKey1(a.data()), Slice(tmp, 1000));
    ASSERT_TRUE(res);

    std::string value;
    Status s;
    ReadOptions read_opts;
    SequenceNumber sq = 0;
    MergeContext merge_context;
    res = mem->Get(LookupKey(genMVCCKey1(a.data()), seq), &value, &s,
                   &merge_context, &sq, &sq, read_opts);
    ASSERT_TRUE(res);
    ASSERT_EQ(value.length(), 1000);
  }

  std::cout << "-----------insert all records ok." << std::endl;

  seq = 10000000;
  for (int i = 0; i < record_sum; i++) {
    seq++;
    std::string a = std::to_string(seq) + "_test";
    std::string value;
    Status s;
    ReadOptions read_opts;
    SequenceNumber sq = 0;
    MergeContext merge_context;
    res = mem->Get(LookupKey(genMVCCKey1(a.data()), seq), &value, &s,
                   &merge_context, &sq, &sq, read_opts);
    ASSERT_TRUE(res);
    ASSERT_EQ(value.length(), 1000);
  }
  std::cout << "-----------get all records ok." << std::endl;
  delete mem;
  std::cout << "-------------------------------------------------------end ..."
            << std::endl;
}

// Deletion and handle for cut
TEST_F(DBMemTableTest, Deletion) {
  Options options;
  // Create a MemTable
  InternalKeyComparator cmp(BytewiseComparator());
  options.memtable_factory = std::make_shared<PureMemFactory>();
  options.pureMemTable = true;
  options.cf_paths.emplace_back(dbname_, 0);
  ImmutableCFOptions ioptions(options);
  WriteBufferManager wb(options.db_write_buffer_size);
  MemTable* mem = new MemTable(cmp, ioptions, MutableCFOptions(options), &wb,
                               kMaxSequenceNumber, 0 /* column_family_id */);
  
  char tmp[1000] = {'0'};
  SequenceNumber seq = 1000000;
  int record_sum = 5000;
  for (int i = 0; i < 100000; i++) {
    std::string a = std::to_string(seq) + "_test";
    std::string b = std::to_string(seq) + "_tast";
    std::string c = std::to_string(seq) + "_trst";
    std::string d = std::to_string(seq) + "_tgst";
    std::string value;
    Status s;
    ReadOptions read_opts;
    SequenceNumber sq = 0, sq1 = 0;
    MergeContext merge_context;

    bool res = mem->Add(seq++, kTypeValue, genMVCCKey1(a.data()), Slice(tmp, 1000));
    res = mem->Add(seq++, kTypeDeletion, genMVCCKey1(a.data()), Slice(tmp, 1000));
    res = mem->Add(seq++, kTypeValue, genMVCCKey1(a.data()), Slice(tmp, 1000));
    res = mem->Add(seq++, kTypeDeletion, genMVCCKey1(a.data()), Slice(tmp, 1000));
    res = mem->Add(seq++, kTypeValue, genMVCCKey1(b.data()), Slice(tmp, 1000));
    res = mem->Add(seq++, kTypeValue, genMVCCKey1(c.data()), Slice(tmp, 1000));
    res = mem->Add(seq++, kTypeValue, genMVCCKey1(a.data()), Slice(tmp, 1000));
    res = mem->Add(seq++, kTypeValue, genMVCCKey1(b.data()), Slice(tmp, 1000));
    assert(res);
    res = mem->Add(seq++, kTypeValue, genMVCCKey1(a.data()), Slice(tmp, 1000));
    assert(res);
    res = mem->Add(seq++, kTypeValue, genMVCCKey1(b.data()), Slice(tmp, 1000));
    assert(res);
  }
  std::cout << "-----------insert all records ok." << std::endl;
  delete mem;
  std::cout << "-------------------------------------------------------end ..."
            << std::endl;
}

// Deletion
TEST_F(DBMemTableTest, DeletionWrongCase1) {
  Options options;
  // Create a MemTable
  InternalKeyComparator cmp(BytewiseComparator());
  options.memtable_factory = std::make_shared<PureMemFactory>();
  options.pureMemTable = true;
  options.cf_paths.emplace_back(dbname_, 0);
  ImmutableCFOptions ioptions(options);
  WriteBufferManager wb(options.db_write_buffer_size);
  MemTable* mem = new MemTable(cmp, ioptions, MutableCFOptions(options), &wb,
                               kMaxSequenceNumber, 0 /* column_family_id */);
  
  char tmp[1000] = {'0'};
  SequenceNumber seq = 10000;
  for (int i = 0; i < 8000; i++) {
    std::string a = std::to_string(seq) + "_1";
    std::string b = std::to_string(seq) + "_2";
    std::string c = std::to_string(seq) + "_3";
    std::string d = std::to_string(seq) + "_4";
    std::string value;
    Status s;
    ReadOptions read_opts;
    SequenceNumber sq = 0, sq1 = 0;
    MergeContext merge_context;

    mem->Add(seq++, kTypeValue, genMVCCKey1(a.data()), Slice(tmp, 1000));
    mem->Add(seq++, kTypeValue, genMVCCKey1(a.data()), Slice(tmp, 1000));

    mem->Add(seq++, kTypeValue, genMVCCKey1(b.data()), Slice(tmp, 1000));
    mem->Add(seq++, kTypeValue, genMVCCKey1(b.data()), Slice(tmp, 1000));
    mem->Add(seq++, kTypeDeletion, genMVCCKey1(b.data()), Slice(tmp, 1000));

    mem->Add(seq++, kTypeValue, genMVCCKey1(c.data()), Slice(tmp, 1000));
    mem->Add(seq++, kTypeValue, genMVCCKey1(a.data()), Slice(tmp, 1000));
  }

  delete mem;
  std::cout << "---------------------------------------------end ..." << std::endl;
}

TEST_F(DBMemTableTest, RangeDeletion) {
  Options options;
  // Create a MemTable
  InternalKeyComparator cmp(BytewiseComparator());
  options.memtable_factory = std::make_shared<PureMemFactory>();
  options.pureMemTable = true;
  options.cf_paths.emplace_back(dbname_, 0);
  ImmutableCFOptions ioptions(options);
  WriteBufferManager wb(options.db_write_buffer_size);
  MemTable* mem = new MemTable(cmp, ioptions, MutableCFOptions(options), &wb,
                               kMaxSequenceNumber, 0 /* column_family_id */);
  std::cout << "memtable create ok." << std::endl;
  // Write some keys and make sure it returns false on duplicates
  bool res;
  std::cout << "Enter the loop" << std::endl;
  // Test the duplicate keys under stress
  char tmp[1000] = {'0'};
  SequenceNumber seq = 100000;
  std::string value;
  Status s;
  ReadOptions read_opts;
  SequenceNumber sq = 0;
  MergeContext merge_context;

  for (int i = 0; i < 90000; i++) {
    seq++;
    std::string a = std::to_string(seq) + "_tast";
    std::string b = std::to_string(seq) + "_tbst";
    std::string c = std::to_string(seq) + "_tcst";
    std::string d = std::to_string(seq) + "_tdst";
    std::string e = std::to_string(seq) + "_test";
    std::string f = std::to_string(seq) + "_tfst";
    std::string g = std::to_string(seq) + "_tgst";
    std::string h = std::to_string(seq) + "_thst";
    res = mem->Add(seq++, kTypeValue, genMVCCKey1(a.data()), Slice(tmp, 1000));
    res = mem->Add(seq++, kTypeValue, genMVCCKey1(b.data()), Slice(tmp, 1000));
    res = mem->Add(seq++, kTypeValue, genMVCCKey1(c.data()), Slice(tmp, 1000));
    res = mem->Add(seq++, kTypeValue, genMVCCKey1(d.data()), Slice(tmp, 1000));
    res = mem->Add(seq++, kTypeValue, genMVCCKey1(e.data()), Slice(tmp, 1000));
    res = mem->Add(seq++, kTypeValue, genMVCCKey1(f.data()), Slice(tmp, 1000));
    res = mem->Add(seq++, kTypeValue, genMVCCKey1(g.data()), Slice(tmp, 1000));
    res = mem->Add(seq++, kTypeValue, genMVCCKey1(h.data()), Slice(tmp, 1000));

    res = mem->Add(seq++, kTypeRangeDeletion, genMVCCKey1(a.data()), Slice(c));
    res = mem->Add(seq++, kTypeValue, genMVCCKey1(a.data()), Slice(tmp, 1000));
    ASSERT_TRUE(res);
    res = mem->Add(seq++, kTypeValue, genMVCCKey1(b.data()), Slice(tmp, 1000));
    res = mem->Add(seq++, kTypeValue, genMVCCKey1(a.data()), Slice(tmp, 1000));
    res = mem->Add(seq++, kTypeValue, genMVCCKey1(c.data()), Slice(tmp, 1000));
    res = mem->Get(LookupKey(genMVCCKey1(a.data()), seq - 13), &value, &s,
                   &merge_context, &sq, &sq, read_opts);
    ASSERT_FALSE(res);

    res = mem->Get(LookupKey(genMVCCKey1(b.data()), seq - 12), &value, &s,
                   &merge_context, &sq, &sq, read_opts);
    ASSERT_FALSE(res);

    res = mem->Get(LookupKey(genMVCCKey1(a.data()), seq - 4), &value, &s,
                   &merge_context, &sq, &sq, read_opts);
    ASSERT_TRUE(res);

    res = mem->Get(LookupKey(genMVCCKey1(d.data()), seq - 10), &value, &s,
                   &merge_context, &sq, &sq, read_opts);
    ASSERT_TRUE(res);
  }

  delete mem;
  std::cout << "---------------------------------------------------end " << std::endl;
}

}  // namespace rocksdb

int main(int argc, char** argv) {
  rocksdb::port::InstallStackTraceHandler();
  ::testing::InitGoogleTest(&argc, argv);
  return RUN_ALL_TESTS();
}
